/**
 * Copyright (C) 2024-2030 Super BilGen & 超亿时代信息技术股份有限公司.All Rights Reserved.
 */
package cn.com.yang.file;


import java.io.BufferedOutputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.CharsetDecoder;
import java.nio.charset.CodingErrorAction;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicLong;
import java.nio.charset.*;
/**
 * @Model cn.com.yang.thread.IndustrialGradeLineProcessor
 * @Desc  TODO
 * @author yangyijin
 * @since 2025年5月22日-上午10:04:27
 */
public class IndustrialGradeLineProcessor {
	private static final Charset UTF8 = StandardCharsets.UTF_8;
    // 性能调优参数
	 // 配置参数
    private static final int THREADS = Runtime.getRuntime().availableProcessors();
    private static final int INDEX_BUFFER = 64 * 1024; // 64KB索引缓冲区
    private static final int READ_BUFFER = 2 * 1024 * 1024; // 2MB读取缓冲区
    
    // 全局进度监控
    private static final AtomicLong processedLines = new AtomicLong(0);

    public static void main(String[] args) throws Exception {
        final String inputFile = "F:\\20gb_data.txt";
        final String indexFile = "F:\\20gb_index.idx";
        
        // 阶段1：构建行索引（可复用）
        if (!new File(indexFile).exists()) {
            buildLineIndex(inputFile, indexFile);
        }
        
        // 阶段2：并行处理
        processParallel(inputFile, indexFile);
    }



    /**
     * 构建行偏移量索引文件
     */
    private static void buildLineIndex(String dataFile, String indexFile) throws IOException {
        try (RandomAccessFile raf = new RandomAccessFile(dataFile, "r");
             FileChannel channel = raf.getChannel();
             DataOutputStream indexOut = new DataOutputStream(
                 new BufferedOutputStream(Files.newOutputStream(Paths.get(indexFile))))) {

            ByteBuffer buffer = ByteBuffer.allocate(INDEX_BUFFER);
            long position = 0;
            boolean newLine = true;

            while (channel.read(buffer) != -1) {
                buffer.flip();
                while (buffer.hasRemaining()) {
                    byte b = buffer.get();
                    if (newLine) {
                        indexOut.writeLong(position);
                        newLine = false;
                    }
                    if (b == '\n'||b == '\r') {
                        newLine = true;
                    }
                    position++;
                }
                buffer.clear();
            }
        }
    }

    /**
     * 并行处理主逻辑
     */
    private static void processParallel(String dataFile, String indexFile) throws Exception {
        // 读取索引文件获取总行数
        long totalLines = Files.size(Paths.get(indexFile)) / Long.BYTES;
        long linesPerThread = (totalLines + THREADS - 1) / THREADS;

        ExecutorService executor = Executors.newFixedThreadPool(THREADS);
        List<Future<?>> futures = new ArrayList<>();

        try (RandomAccessFile indexRaf = new RandomAccessFile(indexFile, "r")) {
            for (int i = 0; i < THREADS; i++) {
                final long startLine = i * linesPerThread;
                final long endLine = Math.min(startLine + linesPerThread, totalLines);

                // 获取实际文件偏移量
                long startOffset = getLineOffset(indexRaf, startLine);
                long endOffset = (endLine >= totalLines) ? 
                    Files.size(Paths.get(dataFile)) : getLineOffset(indexRaf, endLine);

                futures.add(executor.submit(() -> 
                    processBlock(dataFile, startOffset, endOffset, startLine, endLine)
                ));
            }
        }

        // 等待所有任务完成
        for (Future<?> f : futures) {
            f.get();
        }
        executor.shutdown();
    }

    /**
     * 获取指定行的文件偏移量
     */
    private static long getLineOffset(RandomAccessFile indexRaf, long line) throws IOException {
        indexRaf.seek(line * Long.BYTES);
        return indexRaf.readLong();
    }

    /**
     * 处理指定文件块
     */
    private static void processBlock(String dataFile, long start, long end, 
                                    long startLine, long endLine) {
        try (RandomAccessFile raf = new RandomAccessFile(dataFile, "r");
             FileChannel channel = raf.getChannel()) {

            ByteBuffer buffer = ByteBuffer.allocate(READ_BUFFER);
            channel.position(start);
//            CharsetDecoder decoder = UTF8.newDecoder()
//                    .onMalformedInput(CodingErrorAction.REPLACE)
//                    .onUnmappableCharacter(CodingErrorAction.REPLACE);
            StringBuilder sb = new StringBuilder(1024);
            long currentLine = startLine;

            while (channel.position() < end && currentLine < endLine) {
                buffer.clear();
                channel.read(buffer);
                buffer.flip();
                
                while (buffer.hasRemaining() && currentLine < endLine) {
                    byte b = buffer.get();
                    if (b == '\n') {
                        // 处理完整行
                        processLine(new String(sb.toString().getBytes(),"UTF-8"), currentLine);
                        sb.setLength(0);
                        currentLine++;
                    } else if (b != '\r') { // 跳过Windows换行符
                        sb.append((char) b);
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 示例行处理逻辑
     */
    private static void processLine(String line, long lineNumber) {
    	System.out.println("lineNumber-"+lineNumber +":line >" +line);
        // 示例：每处理100万行打印进度 1_000_000
        if (lineNumber %  100 == 0) {
            System.out.printf("Thread %s processed line %,d%n", 
                Thread.currentThread().getName(), lineNumber);
        }
    }
}
